//
//  @ Project : Galaxy Star Trading System
//  @ File Name : MsgQueue.cpp
//  @ Date : 2014/9/5
//  @ Author :
//
//
#include "../StdInc.h"

namespace GSTrader
{
	namespace Util
	{
		CMsgInfo::CMsgInfo()
		{
			m_nPlatfromType = -1;
			strcpy(m_szInvestorId, "");
		}

		CMsgInfo::~CMsgInfo()
		{

		}

		CMsg::CMsg()
		{
			m_bDestoryMessage = false;
			m_bAutoDelete = true;
			m_nType = 0;

			m_pData = NULL;
			m_pExtData = NULL;

			m_nLen = 0;
			m_nLenExt = 0;
			m_bIsLast = false;
		}

		CMsg::CMsg(int nType, int nSize, bool bDestory, int nExtSize)
		{
			m_bDestoryMessage = bDestory;
			m_bAutoDelete = true;
			m_bIsLast = false;
			m_nType = nType;

			m_time = system_clock::now();
			m_pData = new char[nSize];
			m_nLen = nSize;
			if (nExtSize > 0)
			{
				m_pExtData = new char[nExtSize];
			}
			else
			{
				m_pExtData = NULL;
			}
			m_nLenExt = nExtSize;
		}

		CMsg::CMsg(const CMsg& r)
		{
			(*this) = r;
		}

		CMsg& CMsg::operator = (const CMsg& r)
		{
			m_bDestoryMessage = r.m_bDestoryMessage;
			m_bAutoDelete = r.m_bAutoDelete;

			this->m_nType = r.m_nType;
			this->m_time = r.m_time;
			m_nLenExt = r.m_nLenExt;
			m_nLen = r.m_nLen;

			this->m_pData = new char[m_nLen];
			memcpy(m_pData, r.m_pData, m_nLen);

			if (r.m_nLenExt > 0)
			{
				m_pExtData = new char[m_nLenExt];
				memcpy(m_pExtData, r.m_pExtData, m_nLenExt);
			}

			return *this;
		}

		CMsg::~CMsg()
		{
			if (m_bDestoryMessage)
			{
				CDataObject* pDataObject = (CDataObject*)GetData();
				pDataObject->DeleteSelf();
			}

			_delarr(m_pData)
			_delarr(m_pExtData)
		}

		char* CMsg::GetData()
		{
			return m_pData;
		}

		char* CMsg::GetExtData()
		{
			return m_pExtData;
		}

		void CMsg::SetTime(CTimePoint t)
		{
			m_time = t;
		}

		CMsgFilter::CMsgFilter()
		{

		}

		CMsgFilter::~CMsgFilter()
		{

		}

		CMsgQueue::CMsgQueue() : m_nEmptyCount(0), m_nPopCount(0), m_nPushCount(0), m_nFullCount(0)
		{

		}

		CMsgQueue::~CMsgQueue()
		{

		}

		void CMsgQueue::Push(CMsg* p)
		{
			lock_guard<mutex> lk(m_Mutex);
			m_Queue.push(p);
			m_nPushCount++;
			m_condition.notify_one();
		}

		size_t CMsgQueue::GetSize()
		{
			return m_Queue.size();
		}

		CMsg* CMsgQueue::Pop()
		{
			lock_guard<mutex> lk(m_Mutex);
			if (m_Queue.empty())
			{
				m_nEmptyCount++;
				return nullptr;
			}
			CMsg* pNewMsg = m_Queue.front();
			m_Queue.pop();
			m_nPopCount++;
			return pNewMsg;
		}

		CMsg* CMsgQueue::WaitPop()
		{
			unique_lock<mutex> lk(m_Mutex);
			m_condition.wait(lk, [this] {return !m_Queue.empty();});

			CMsg* newMsg = (m_Queue.front());
			m_Queue.pop();
			m_nPopCount++;

			return newMsg;
		}

		CMsg* CMsgQueue::GetTopMsg()
		{
			lock_guard<mutex> lk(m_Mutex);
			if (m_Queue.empty())
				return nullptr;

			CMsg* newMsg = m_Queue.front();
			return newMsg;
		}

		bool CMsgQueue::Empty()
		{
			lock_guard<mutex> lk(m_Mutex);
			return m_Queue.empty();
		}

		void CMsgQueue::Clear()
		{
			lock_guard<mutex> lk(m_Mutex);
			while (!m_Queue.empty())
				m_Queue.pop();
		}

		CSortMsgQueue::CSortMsgQueue()
		{
			m_Queue.reserve(100);
		}

		CSortMsgQueue::~CSortMsgQueue()
		{

		}

		CTimePoint CSortMsgQueue::GetTimePoint()
		{
			return m_time;
		}

		int CSortMsgQueue::GetIndex(CMsg* pMsg)
		{
			lock_guard<recursive_mutex> lock(m_Mutex);

			int nSize = m_Queue.size();
			if (nSize == 0)
				return 0;

			CMsg* pTheMsg = m_Queue[0];
			if (pTheMsg->GetTime() > pMsg->GetTime())
			{
				return 0;
			}

			pTheMsg = m_Queue[nSize - 1];
			if (pTheMsg->GetTime() <= pMsg->GetTime())
			{
				return m_Queue.size();
			}

			int nLow = 0;
			int nMid = 0;
			int nHigh = nSize;
			while (nLow != nHigh)
			{
				nMid = ((nLow + nHigh) + 1) / 2;
				if (m_Queue[nMid]->GetTime() <= pMsg->GetTime())
				{
					if (m_Queue[nMid + 1]->GetTime() > pMsg->GetTime())
					{
						return (nMid + 1);
					}

					nLow = nMid;
				}
				else
				{
					nHigh = nLow - 1;
				}

				if (nLow > nHigh)
				{
					assert(false);
					break;
				}
			}

			return (nLow + 1);
		}

		void CSortMsgQueue::Push(CMsg* pMsg)
		{
			lock_guard<recursive_mutex> lock(m_Mutex);

			int nIndex = GetIndex(pMsg);
			assert(nIndex >= 0);

			m_Queue.insert(m_Queue.begin() + nIndex, pMsg);
			m_time = m_Queue[0]->GetTime();
		}

		CMsg* CSortMsgQueue::Pop()
		{
			lock_guard<recursive_mutex> lock(m_Mutex);

			if (m_Queue.empty())
				return nullptr;

			CMsg* pMsg = m_Queue[0];
			m_Queue.erase(m_Queue.begin());
			if (m_Queue.size() > 0)
			{
				m_time = m_Queue[0]->GetTime();
			}

			return pMsg;
		}

		bool CSortMsgQueue::Empty()
		{
			lock_guard<recursive_mutex> lock(m_Mutex);
			return m_Queue.size() == 0;
		}

		void CSortMsgQueue::Clear()
		{
			lock_guard<recursive_mutex> lock(m_Mutex);
			m_Queue.clear();
		}

		CMsg* CSortMsgQueue::GetTopMsg()
		{
			lock_guard<recursive_mutex> lock(m_Mutex);
			if (m_Queue.empty())
				return nullptr;

			CMsg* pMsg = m_Queue[0];
			return pMsg;
		}

		CMsgPipe::CMsgPipe() : m_nPopIndex(0)
		{

		}

		CMsgPipe::~CMsgPipe()
		{

		}

		void CMsgPipe::Clear()
		{
			for (auto p : m_vMsgQueue)
			{
				p->Clear();
			}
		}

		int CMsgPipe::GetQueueCount()
		{
			return m_vMsgQueue.size();
		}

		CMsgQueue* CMsgPipe::GetQueueByIndex(int nIdx)
		{
			if (nIdx < 0 || nIdx >= (int)m_vMsgQueue.size())
				return nullptr;

			return m_vMsgQueue[nIdx];
		}

		void CMsgPipe::DeleteMsgQueye(CMsgQueue* pMsgQueue)
		{
			lock_guard<mutex> lk(m_Mutex);
			for (auto it = m_vMsgQueue.begin(); it != m_vMsgQueue.end(); it++)
			{
				if ((*it) == pMsgQueue)
				{
					m_vMsgQueue.erase(it);
					return;
				}
			}
		}

		void CMsgPipe::AddMsgQueue(CMsgQueue* pMsgQueue)
		{
			lock_guard<mutex> lk(m_Mutex);
			m_vMsgQueue.push_back(pMsgQueue);
		}

		bool CMsgPipe::IsEmpty()
		{
			lock_guard<mutex> lk(m_Mutex);
			for (auto p : m_vMsgQueue)
			{
				if (p == NULL)
				{
					assert(false);
					continue;
				}
				if (!p->Empty())
				{
					return false;
				}
			}

			return true;
		}

		CMsg* CMsgPipe::Pop()
		{
			lock_guard<mutex> lk(m_Mutex);

			size_t nSize = m_vMsgQueue.size();
			if (nSize == 0)
				return nullptr;

			CTimePoint tmMin = CTimePoint::Now();

			CMsgQueue* pMintimeMsgQueue = nullptr;
			for (auto p : m_vMsgQueue)
			{
				CMsg* pBody = p->GetTopMsg();
				if (pBody != nullptr && pBody->GetTime() <= tmMin)
				{
					tmMin = pBody->GetTime();
					pMintimeMsgQueue = p;
				}
			}

			if (pMintimeMsgQueue)
			{
				return pMintimeMsgQueue->Pop();
			}

			return nullptr;
		}

		CMsgLogger::CMsgLogger()
		{
		}

		CMsgLogger::~CMsgLogger()
		{
		}
	}
}